package com.caltco.cargo.ops.common.utils;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.UUID;

/**
 * @Author Bernie
 * @Date 2020/7/15 11:20
 * RabbitMQ 工具类
 */
@Slf4j
@Component
@ConditionalOnProperty(prefix = "spring.rabbitmq", name = "enabled", havingValue = "true")
public class RabbitMQUtil {

    private final AmqpAdmin amqpAdmin;

    private final RabbitTemplate rabbitTemplate;

    /** 自定义的Exchange */
    private final TopicExchange topicExchange;


    @Autowired
    public RabbitMQUtil(AmqpAdmin amqpAdmin, RabbitTemplate rabbitTemplate,TopicExchange topicExchange){
        this.amqpAdmin = amqpAdmin;
        this.rabbitTemplate = rabbitTemplate;
        this.topicExchange = topicExchange;
    }


    /**
     * 没有绑定KEY的Exchange发送
     * @param exchange
     * @param msg
     */
    public void sendMessageToExchange(AbstractExchange exchange, String msg){
        this.addExchange(exchange);
        log.info("RabbitMQ send {} -> {}",exchange.getName(),msg);
        rabbitTemplate.convertAndSend(topicExchange.getName(),msg);
    }

    /**
     * 给queue发送消息
     * @param queueName
     * @param msg
     */
    @Async
    public void sendToQueue(String queueName,String msg){
        Queue queue = new Queue(queueName);
        //this.addQueue(queue);
        Binding binding = BindingBuilder.bind(queue).to(DirectExchange.DEFAULT).withQueueName();
        amqpAdmin.declareBinding(binding);
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);

        Message message = MessageBuilder.withBody(msg.getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_JSON)
                .setContentEncoding("base64")
                .setMessageId(UUID.randomUUID()+"")
                .build();

        log.info("发送的消息体[{}]", message.toString());
        rabbitTemplate.convertAndSend(DirectExchange.DEFAULT.getName(),queueName,message);
    }

    /**
     * 通过路由向指定的交换器返送消息
     * @param exchangeName 交换器名称
     * @param routingKey 路由
     * @param msg 消息，最好是json字符串
     */
    @Async
    public void sendToExchange(String exchangeName, String routingKey, String msg) {

        Message message = MessageBuilder.withBody(msg.getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_JSON)
                .setContentEncoding("base64")
                .setMessageId(UUID.randomUUID()+"")
                .build();
        rabbitTemplate.convertAndSend(exchangeName,routingKey,message);
    }
    @Async
    public void sendToExchange2(String exchangeName, String routingKey, Map<String,Object> msg) {
        rabbitTemplate.convertAndSend(exchangeName,routingKey,msg,new CorrelationData(UUID.randomUUID().toString()));
    }

    /**
     * 创建Exchange
     * @param exchange
     */
    private void addExchange(AbstractExchange exchange){
        amqpAdmin.declareExchange(exchange);
    }

    /**
     * 删除一个Exchange
     * @param exchangeName
     */
    public boolean deleteExchange(String exchangeName){
        return amqpAdmin.deleteExchange(exchangeName);
    }


    /**
     * Declare a queue whose name is automatically named. It is created with exclusive = true, autoDelete=true, and
     * durable = false.
     * @return Queue
     */
    public Queue addQueue(){
        return amqpAdmin.declareQueue();
    }

    /**
     * 创建一个指定的Queue
     * @param queue
     * @return queueName
     */
    public String addQueue(Queue queue){
        return amqpAdmin.declareQueue(queue);
    }

    /**
     * 创建一个绑定交换器和指定路由的队列
     * @param queue 目标队列
     * @param exchangeName 交换器名称
     * @param routingKey 路由
     */
    public void addQueue(Queue queue, String exchangeName,String routingKey) {
        addQueue(queue);
        addBinding(queue, new TopicExchange(exchangeName),routingKey);
    }

    /**
     * Delete a queue.
     * @param queueName the name of the queue.
     * @param unused true if the queue should be deleted only if not in use.
     * @param empty true if the queue should be deleted only if empty.
     */
    public void deleteQueue(String queueName, boolean unused, boolean empty){
        amqpAdmin.deleteQueue(queueName,unused,empty);
    }

    /**
     * 删除一个queue
     * @return true if the queue existed and was deleted.
     * @param queueName
     */
    public boolean deleteQueue(String queueName){
        return amqpAdmin.deleteQueue(queueName);
    }

    /**
     * 绑定一个队列到一个匹配型交换器使用一个routingKey
     * @param queue
     * @param exchange
     * @param routingKey
     */
    public void addBinding(Queue queue ,TopicExchange exchange,String routingKey){
        Binding binding = BindingBuilder.bind(queue).to(exchange).with(routingKey);
        amqpAdmin.declareBinding(binding);
    }

    /**
     * 绑定一个Exchange到一个匹配型Exchange 使用一个routingKey
     * @param exchange
     * @param topicExchange
     * @param routingKey
     */
    public void addBinding(Exchange exchange,TopicExchange topicExchange,String routingKey){
        Binding binding = BindingBuilder.bind(exchange).to(topicExchange).with(routingKey);
        amqpAdmin.declareBinding(binding);
    }
    /**
     * 去掉一个binding
     * @param binding
     */
    public void removeBinding(Binding binding){
        amqpAdmin.removeBinding(binding);
    }
}
